package org.nbict.iot.trash.trident;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.trident.Stream;
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.trident.operation.BaseFunction;
import org.apache.storm.trident.operation.TridentCollector;
import org.apache.storm.trident.testing.FixedBatchSpout;
import org.apache.storm.trident.tuple.TridentTuple;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.nbict.iot.trident.mogo.MultiQueryFilterCreator;
import org.nbict.iot.trident.mogo.mapper.MongoMapper;
import org.nbict.iot.trident.mogo.mapper.impl.SimpleMongoUpdateMapper;
import org.nbict.iot.trident.mogo.state.MongoState;
import org.nbict.iot.trident.mogo.state.MongoStateFactory;
import org.nbict.iot.trident.mogo.state.MongoStateUpdater;

/**
 * Created by songseven on 18/6/25.
 */
@Deprecated
public class RealtimeCarInfoTrident {

    private static final String url = "127.0.0.1:27011";
    private static final String database = "iot_car_test";
    private static final String username = "nbicc";
    private static final String password = "43b5fda47f34442091351698b9136e87";
    private static final String admin = "admin";
    private static final String collectionName = "test-mongo";


    public static StormTopology mongoTrident(){

        TridentTopology topology = new TridentTopology();

        Fields fields = new Fields("word", "count");

        FixedBatchSpout spout = new FixedBatchSpout(fields, 2,
                new Values("storm", 2),
                new Values("trident", 3),
                new Values("needs", 4),
                new Values("javadoc", 5)
        );

        spout.setCycle(true);

        Stream stream = topology.newStream("spout1", spout);

        //MongoMapper mapper = new SimpleMongoMapper().withFields("word", "count");
//        MongoMapper mapper = new SimpleMongoUpdateMapper().withFields("word", "count")
//                .withFilter(new SimpleQueryFilterCreator().withField("word")); //过滤条件

        MongoMapper mapper = new SimpleMongoUpdateMapper().withFields("word", "count")
                .withFilter(new MultiQueryFilterCreator().eq("word")); //过滤条件

        MongoState.Options options = new MongoState.Options()
                .withUrl(url)
                .withDatabase(database)
                .withUsername(username)
                .withPassword(password)
                .withAdmin(admin)
                .withCollectionName(collectionName)
                .withMapper(mapper);

        stream.partitionPersist( new MongoStateFactory(options), fields,
                new MongoStateUpdater(), new Fields());

        stream.each(new Fields("word", "count"), new BaseFunction() {
            @Override
            public void execute(TridentTuple tuple, TridentCollector
                    collector) {
                    System.out.println(tuple);
            }
        }, new Fields());

        return topology.build();
    }


    public static void main(String[] args) {
        Config conf = new Config();
        conf.setMaxSpoutPending(3);
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("test", conf,  mongoTrident());

        try {
            Thread.sleep(100000);
            cluster.shutdown();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
